package com.jcode.netty.core;

import com.jcode.netty.exception.RemotingSendRequestException;
import com.jcode.netty.exception.RemotingTimeoutException;
import com.jcode.netty.exception.RemotingTooMuchRequestException;
import com.jcode.netty.cmd.SysResponseCode;
import com.jcode.netty.protocol.Packet;
import com.jcode.netty.protocol.PacketUtil;
import com.jcode.netty.util.RemotingHelper;
import com.jcode.netty.util.ServiceThread;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import org.apache.commons.lang3.tuple.MutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.concurrent.*;

/**
 * 远程服务抽象实现
 *
 * @Author dreajay
 * @DateTime 2016年12月6日 下午2:28:30
 *
 */
public abstract class AbstractRemotingService implements RemotingService {

	protected final Semaphore semaphoreOneway;
	protected final Semaphore semaphoreAsync;
	/** 处理码-处理器 */
	protected final HashMap<Integer, Pair<RequestProcessor, ExecutorService>> processorTable = new HashMap<Integer, Pair<RequestProcessor, ExecutorService>>(64);

	/** 请求响应表 */
	protected final ConcurrentHashMap<Long, ResponseFuture> responseTable = new ConcurrentHashMap<Long, ResponseFuture>(256);

	/** 默认线程池 */
	protected ExecutorService defaultRequestExecutor;
	/** 默认处理器 */
	protected Pair<RequestProcessor, ExecutorService> defaultRequestProcessor;

	/** 回调线程池 */
	protected ExecutorService callbackExecutor;

	/** 通道IO事件监听器 */
	protected ChannelEventListener channelEventListener;

	/** 网络事件队列最大值 */
	private final int maxEventSize = 100000;

	/** 网络事件队列 */
	private final LinkedBlockingQueue<NetEvent> eventQueue = new LinkedBlockingQueue<NetEvent>(maxEventSize);

	protected final NetEventThreadExecuter eventExecuter = new NetEventThreadExecuter();

	private static final Logger logger = LoggerFactory.getLogger(AbstractRemotingService.class);

	public AbstractRemotingService(final int permitsOneway, final int permitsAsync) {
		this.semaphoreOneway = new Semaphore(permitsOneway, true);
		this.semaphoreAsync = new Semaphore(permitsAsync, true);
	}

	public AbstractRemotingService(final int permitsOneway, final int permitsAsync, ChannelEventListener channelEventListener) {
		this.semaphoreOneway = new Semaphore(permitsOneway, true);
		this.semaphoreAsync = new Semaphore(permitsAsync, true);
		this.channelEventListener = channelEventListener;
	}

	@Override
	public void start() {
		eventExecuter.start();
	}

	@Override
	public void shutdown() {
		if (defaultRequestExecutor != null) {
			defaultRequestExecutor.shutdown();
		}
		if (callbackExecutor != null) {
			callbackExecutor.shutdown();
		}
		eventExecuter.shutdown();
	}

	public ChannelEventListener getChannelEventListener() {
		return channelEventListener;
	}

	public void setChannelEventListener(ChannelEventListener channelEventListener) {
		this.channelEventListener = channelEventListener;
	}

	@Override
	public ExecutorService getCallbackExecutor() {
		return callbackExecutor;
	}

	public void setCallbackExecutor(ExecutorService callbackExecutor) {
		this.callbackExecutor = callbackExecutor;
	}

	@Override
	public ExecutorService getDefaultRequestExecutor() {
		return this.defaultRequestExecutor;
	}

	/**
	 * 设置默认处理器，默认请求处理线程
	 * 
	 * @param processor
	 * @param defaultRequestExecutor
	 */
	public void registerDefaultProcessor(final RequestProcessor processor, final ExecutorService defaultRequestExecutor) {
		this.defaultRequestExecutor = defaultRequestExecutor;
		this.defaultRequestProcessor = new MutablePair<RequestProcessor, ExecutorService>(processor, defaultRequestExecutor);
	}

	@Override
	public void registerProcessor(int requestCode, RequestProcessor processor, ExecutorService executor) {
		ExecutorService executorThis = executor;
		if (null == executor) {
			executorThis = getDefaultRequestExecutor();
		}
		Pair<RequestProcessor, ExecutorService> pair = new MutablePair<RequestProcessor, ExecutorService>(processor, executorThis);
		this.processorTable.put(requestCode, pair);
	}

	@Override
	public Packet sendRequestSync(final Channel channel, final Packet packet, long timeoutMillis) throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException {
		try {
			final ResponseFuture responseFuture = new ResponseFuture(packet.getSequenceId(), timeoutMillis, null, null);
			// 放入等待响应的table中
			this.responseTable.put(packet.getSequenceId(), responseFuture);
			channel.writeAndFlush(packet).addListener(new ChannelFutureListener() {
				@Override
				public void operationComplete(ChannelFuture f) throws Exception {
					if (f.isSuccess()) {
						responseFuture.setSendRequestOK(true);
						return;
					}
					// 失败
					responseFuture.setSendRequestOK(false);
					responseTable.remove(packet.getSequenceId());
					responseFuture.setCause(f.cause());
					responseFuture.putResponse(null);
					logger.warn("send a request command to channel <" + channel.remoteAddress() + "> failed.");
				}
			});

			Packet responsePacket = responseFuture.waitResponse(timeoutMillis);
			if (null == responsePacket) {
				if (responseFuture.isSendRequestOK()) {
					// 网络超时
					throw new RemotingTimeoutException(RemotingHelper.parseChannelRemoteAddr(channel), timeoutMillis, responseFuture.getCause());
				} else {
					// 发送异常
					throw new RemotingSendRequestException(RemotingHelper.parseChannelRemoteAddr(channel), responseFuture.getCause());
				}
			}
			return responsePacket;
		} finally {
			responseTable.remove(packet.getSequenceId());
		}
	}

	@Override
	public void sendRequestAsync(final Channel channel, final Packet packet, long timeoutMillis, ResponseCallback callback) throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException, RemotingTooMuchRequestException {
		boolean acquired = this.semaphoreAsync.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
		if (acquired) {
			final SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce(this.semaphoreAsync);
			final ResponseFuture responseFuture = new ResponseFuture(packet.getSequenceId(), timeoutMillis, callback, once);
			this.responseTable.put(packet.getSequenceId(), responseFuture);
			try {
				channel.writeAndFlush(packet).addListener(new ChannelFutureListener() {
					@Override
					public void operationComplete(ChannelFuture f) throws Exception {
						if (f.isSuccess()) {
							responseFuture.setSendRequestOK(true);
							return;
						}
						responseFuture.setSendRequestOK(false);
						responseFuture.putResponse(null);
						responseTable.remove(packet.getSequenceId());
						// 执行回调
						try {
							responseFuture.executeInvokeCallback();
						} catch (Throwable e) {
							logger.warn("excute callback in writeAndFlush addListener, and callback throw", e);
						} finally {
							responseFuture.release();
						}

						logger.warn("send a request command to channel <{}> failed.", RemotingHelper.parseChannelRemoteAddr(channel));
					}
				});
			} catch (Exception e) {
				responseFuture.release();
				logger.warn("send a request command to channel <" + RemotingHelper.parseChannelRemoteAddr(channel) + "> Exception", e);
				throw new RemotingSendRequestException(RemotingHelper.parseChannelRemoteAddr(channel), e);
			}
		} else {
			if (timeoutMillis <= 0) {
				throw new RemotingTooMuchRequestException("the request is too fast");
			} else {
				String info = String.format("invokeAsyncImpl tryAcquire semaphore timeout, %dms, waiting thread nums: %d semaphoreAsyncValue: %d", //
						timeoutMillis,//
						this.semaphoreAsync.getQueueLength(),//
						this.semaphoreAsync.availablePermits()//
						);
				logger.warn(info);
				throw new RemotingTimeoutException(info);
			}
		}
	}

	@Override
	public void sendRequestOneway(final Channel channel, final Packet packet, long timeoutMillis) throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException, RemotingTooMuchRequestException {
		boolean acquired = this.semaphoreOneway.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
		if (acquired) {
			final SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce(this.semaphoreOneway);
			try {
				packet.markOnewayType();
				channel.writeAndFlush(packet).addListener(new ChannelFutureListener() {
					@Override
					public void operationComplete(ChannelFuture f) throws Exception {
						once.release();
						if (!f.isSuccess()) {
							logger.warn("send a request command to channel <" + channel.remoteAddress() + "> failed.");
						}
					}
				});
			} catch (Exception e) {
				once.release();
				logger.warn("write send a request command to channel <" + channel.remoteAddress() + "> failed.");
				throw new RemotingSendRequestException(RemotingHelper.parseChannelRemoteAddr(channel), e);
			}
		} else {
			if (timeoutMillis <= 0) {
				throw new RemotingTooMuchRequestException("invokeOnewayImpl invoke too fast");
			} else {
				String info = String.format("invokeOnewayImpl tryAcquire semaphore timeout, %dms, waiting thread nums: %d semaphoreAsyncValue: %d", //
						timeoutMillis,//
						this.semaphoreAsync.getQueueLength(),//
						this.semaphoreAsync.availablePermits()//
						);
				logger.warn(info);
				logger.warn(packet.toString());
				throw new RemotingTimeoutException(info);
			}
		}
	}

	@Override
	public void processMessageReceived(ChannelHandlerContext ctx, final Packet packet) {
		if (packet != null) {
			if (packet.isResponseType()) {
				processResponse(ctx, packet);
			} else {
				processRequest(ctx, packet);
			}
		}
	}

	@Override
	public void processRequest(final ChannelHandlerContext ctx, final Packet packet) {
		final Pair<RequestProcessor, ExecutorService> matched = this.processorTable.get(packet.getCmd());
		final Pair<RequestProcessor, ExecutorService> pair = null == matched ? this.defaultRequestProcessor : matched;
		if (pair == null) {
			logger.error(" {} request cmd {} not supported", RemotingHelper.parseChannelRemoteAddr(ctx.channel()), packet.getCmd());
			final Packet response = PacketUtil.createResponsePacket(packet, SysResponseCode.REQUEST_NOT_SUPPORTED);
			ctx.writeAndFlush(response);
			return;
		}
		Runnable run = new Runnable() {
			@Override
			public void run() {
				try {
					// doBeforeRequest
					final Packet response = pair.getLeft().processRequest(ctx, packet);
					// doAfterResponse
					if (!packet.isOnewayType()) {
						if (response != null) {
							try {
								ctx.writeAndFlush(response);
							} catch (Throwable e) {
								logger.error("process request over, but response failed", e);
								logger.error(packet.toString());
								logger.error(response.toString());
							}
						}
					}
				} catch (Throwable e) {
					logger.error("process request exception", e);
					if (!packet.isOnewayType()) {
						final Packet response = PacketUtil.createResponsePacket(packet, SysResponseCode.SYSTEM_ERROR, null);
						response.setSequenceId(packet.getSequenceId());
						ctx.writeAndFlush(response);
					}
				}
			}
		};

		try {
			pair.getRight().submit(run);
		} catch (RejectedExecutionException e) {
			if ((System.currentTimeMillis() % 10000) == 0) {
				logger.warn(RemotingHelper.parseChannelRemoteAddr(ctx.channel()) //
						+ ", too many requests and system thread pool busy, RejectedExecutionException " //
						+ pair.getRight().toString() //
						+ " request code: " + packet.getCmd());
			}
			if (!packet.isOnewayType()) {
				final Packet response = PacketUtil.createResponsePacket(packet, SysResponseCode.SYSTEM_BUSY, null);
				ctx.writeAndFlush(response);
			}
		}

	}

	@Override
	public void processResponse(ChannelHandlerContext ctx, Packet packet) {
		final ResponseFuture responseFuture = responseTable.get(packet.getSequenceId());
		if (responseFuture != null) {
			responseFuture.setResponse(packet);
			responseFuture.release();
			responseTable.remove(packet.getSequenceId());
			if (responseFuture.getResponseCallback() != null) {
				boolean runInThisThread = false;
				ExecutorService executor = this.getCallbackExecutor();
				if (executor != null) {
					try {
						executor.submit(new Runnable() {
							@Override
							public void run() {
								try {
									responseFuture.executeInvokeCallback();
								} catch (Throwable e) {
									logger.warn("excute callback in executor exception, and callback throw", e);
								}
							}
						});
					} catch (Exception e) {
						runInThisThread = true;
						logger.warn("excute callback in executor exception, maybe executor busy", e);
					}
				} else {
					runInThisThread = true;
				}

				if (runInThisThread) {
					try {
						responseFuture.executeInvokeCallback();
					} catch (Throwable e) {
						logger.warn("executeInvokeCallback Exception", e);
					}
				}
			} else {
				responseFuture.putResponse(packet);
			}
		} else {
			logger.warn("receive response, but not matched any request, " + RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
		}

	}

	/**
	 * 触发网络状态事件
	 * 
	 * @param event
	 */
	public void fireNetEvent(final NetEvent event) {
		if (channelEventListener != null) {
			if (!this.eventQueue.offer(event)) {
				logger.error("event queue size[{}] enough, so drop this event {}", this.eventQueue.size(), event.toString());
			}
		}
	}

	/**
	 * 网络状态事件轮询线程
	 *
	 * @Author dreajay
	 * @DateTime 2016年12月14日 下午4:34:04
	 *
	 */
	class NetEventThreadExecuter extends ServiceThread {
		@Override
		public void run() {
			logger.info(this.getServiceName() + " service started");
			ChannelEventListener listener = null;
			while (!this.isStoped()) {
				try {
					NetEvent event = AbstractRemotingService.this.eventQueue.poll(3000, TimeUnit.MILLISECONDS);
					listener = getChannelEventListener();
					if (event != null && listener != null) {
						switch (event.getType()) {
						case ALL_IDLE:
							listener.onChannelIdle(event.getRemoteAddr(), event.getChannel());
							break;
						case CLOSE:
							listener.onChannelClose(event.getRemoteAddr(), event.getChannel());
							break;
						case CONNECT:
							listener.onChannelConnect(event.getRemoteAddr(), event.getChannel());
							break;
						case EXCEPTION:
							listener.onChannelException(event.getRemoteAddr(), event.getChannel());
							break;
						default:
							break;
						}
					}
				} catch (Exception e) {
					logger.warn(this.getServiceName() + " service has exception. ", e);
				}
			}
			logger.info(this.getServiceName() + " service end");
		}

		@Override
		public String getServiceName() {
			return NetEventThreadExecuter.class.getSimpleName();
		}
	}
}
